package com.dahua.messagesyncsh.kafka.runner;

import com.alibaba.fastjson2.JSON;
import com.dahua.messagesyncsh.pojo.T485;
import com.dahua.messagesyncsh.service.ProcessorService;
import com.dahua.messagesyncsh.util.ObjectUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

@Slf4j
@Component
public class ControlHandleRunner implements Runnable {

    @Resource
    private ProcessorService processorService;

    private final List<ConsumerRecord<String, String>> consumerRecords;

    public ControlHandleRunner(ProcessorService  processorService,
            List<ConsumerRecord<String, String>> consumerRecords) {
         this.processorService = processorService;
        this.consumerRecords = consumerRecords;
    }

    @Override
    public void run() {
        this.handle(consumerRecords);
    }

    public void handle(List<ConsumerRecord<String, String>> consumerRecords) {

        List<T485> t485s = new ArrayList<>();

        for (ConsumerRecord<String, String> item : consumerRecords) {
            try {
                if (item.value() == null || item.value().isEmpty()) {
                    continue;
                }
                T485 t485 = JSON.parseObject(item.value(), T485.class);

                if (ObjectUtil.checkNullField(t485)) {
                    log.error("【Control56】：数据不完整{}", t485.toString());
                    continue;
                }

                boolean allZero = true;
                for (Integer[] data : t485.getData()){
                    if (data[1] != 0) {
                        allZero = false;
                        break;
                    }
                }
                if (allZero) {
                    log.error("【Control56】：数据全为0 {}", t485);
                    continue;
                }

                t485s.add(t485);
            } catch (Exception e) {
                log.error("【Control56】：数据转换失败{} \n {}", item.value(), e.getMessage());
            }
        }

        processorService.handleT485(t485s);
    }
}
